package cn.com.dhcc.turbo.service.rmi;

import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.Set;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import cn.com.dhcc.app.core.exception.ServiceException;
import cn.com.dhcc.app.core.handler.threadpoolhandler.ThreadPool;
import cn.com.dhcc.app.core.service.Log;
import cn.com.dhcc.app.core.service.Log.LOGTYPE;
import cn.com.dhcc.app.core.tag.CommonStaticCode.SUCCESSFAIL;
import cn.com.dhcc.app.core.util.BeanMapper;
import cn.com.dhcc.app.core.util.Status;
import cn.com.dhcc.app.pub.core.consts.AppConst;
import cn.com.dhcc.app.pub.core.consts.AppIocBean;
import cn.com.dhcc.app.pub.core.consts.KernelParam;
import cn.com.dhcc.turbo.entity.api.RouteLog;
import cn.com.dhcc.turbo.entity.exch.CrossNetTmpCmData;
import cn.com.dhcc.turbo.entity.exch.QueueMsgWrapper;
import cn.com.dhcc.turbo.entity.exch.QueueMsgWrapper.ExchageLifecycle;
import cn.com.dhcc.turbo.entity.exch.SendCmData;
import cn.com.dhcc.turbo.service.exch.cmdata.CmDataExchangeLifecycle;
import cn.com.dhcc.turbo.service.registry.sync.ISynchronizeHandler;
import cn.com.dhcc.turbo.service.registry.sync.SyncCallBack;


@Service
public class RmiRemoteServiceImpl extends  UnicastRemoteObject implements ISynchronizeHandler{
	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	protected RmiRemoteServiceImpl() throws RemoteException {
		super();
	}

	@SuppressWarnings("deprecation")
	private static Logger logger = Log.getExistLogger(LOGTYPE.TLQ);
	@Autowired
	@Qualifier(AppIocBean.CM_DATA_EXCHANGE_LIFECYCLE_BEAN)
	private CmDataExchangeLifecycle cmDataExchageLifecycle;
	
	public Status consumeMsg(QueueMsgWrapper queueMsgWrapper){
		Status status=Status.getInstance();
		logger.info("JMS接收消息：queueName[" + queueMsgWrapper.getQueueName()+"],ExchageLifecycle["+queueMsgWrapper.getExchageLifecycle().getDesc()+"]");
		 try {
			if (ExchageLifecycle.remoteStore.equals(queueMsgWrapper.getExchageLifecycle())) {
				SendCmData sendCmData = (SendCmData) queueMsgWrapper.getData();
				String msg = "传输管道[RmiTransportPipeline]接收数据成功";
				RouteLog.setRemotePipeline(sendCmData, AppConst.YES, msg, AppConst.dtFormat.format(new Date()));
				status=cmDataExchageLifecycle.remoteStore(sendCmData);
			}else if(ExchageLifecycle.remoteStoreTmp.equals(queueMsgWrapper.getExchageLifecycle())){
				SendCmData sendCmData = (SendCmData) queueMsgWrapper.getData();
				String msg = "传输管道[RmiTransportPipeline]接收数据成功";
				RouteLog.setInnerRecvPipeline(sendCmData, AppConst.YES, msg, AppConst.dtFormat.format(new Date()));
				status=cmDataExchageLifecycle.remoteStore(sendCmData);
			}else
			if (ExchageLifecycle.feedbackSendStatusAndRoutelog.equals(queueMsgWrapper.getExchageLifecycle())) {
				status=cmDataExchageLifecycle.updateSendStatusAndRoutelog((SendCmData) queueMsgWrapper.getData());
			}else
			if (ExchageLifecycle.feedbackSendStatusAndRoutelogToTmp.equals(queueMsgWrapper.getExchageLifecycle())) {
				SendCmData sendCmData = (SendCmData) queueMsgWrapper.getData();
				CrossNetTmpCmData tempData=new CrossNetTmpCmData();
				BeanMapper.copy(sendCmData, tempData);
				status=cmDataExchageLifecycle.updateSendStatusAndRoutelog(tempData);
			}else
			if (ExchageLifecycle.feedbackNotifyStatusAndRoutelogTmp.equals(queueMsgWrapper.getExchageLifecycle())) {
				SendCmData sendCmData = (SendCmData) queueMsgWrapper.getData();
				CrossNetTmpCmData tempData=new CrossNetTmpCmData();
				BeanMapper.copy(sendCmData, tempData);
				status=cmDataExchageLifecycle.updateNotifyStatusAndRoutelog(tempData);
			}else
			if (ExchageLifecycle.feedbackNotifyStatusAndRoutelog.equals(queueMsgWrapper.getExchageLifecycle())) {
				status=cmDataExchageLifecycle.updateNotifyStatusAndRoutelog((SendCmData) queueMsgWrapper.getData());
			}else
			//跨网传输
			if (ExchageLifecycle.gapEntrepot.equals(queueMsgWrapper.getExchageLifecycle())) {
				SendCmData sendCmData = (SendCmData) queueMsgWrapper.getData();
				String msg = "传输管道[RmiTransportPipeline]接收数据成功";
				RouteLog.setOuterRecvPipeline(sendCmData, AppConst.YES, msg, AppConst.dtFormat.format(new Date()));
				status=cmDataExchageLifecycle.outerEntrepotStore(sendCmData);
			}
		} catch (Exception e) {
			status.setMsg("消息消费报错:"+e.getMessage());
			status.setSuccess(false);
		}
	
		return status;
	}

	@Override
	public SyncCallBack receiveData(Object receivedData) throws RemoteException {
		SyncCallBack syncCallBack=new SyncCallBack();
		try {
			syncCallBack.setResult(SUCCESSFAIL.SUCCESS);
			syncCallBack.setMsg("接收成功");
			this.startConsumeMsg(receivedData);
		} catch (ServiceException e) {
			logger.error("JMS消息接收异常", e);
			syncCallBack.setResult(SUCCESSFAIL.FAIL);
			syncCallBack.setMsg("JMS消息接收异常"+e);
		}
		
		return syncCallBack;
	}

	
	static{
		//创建线程池
		ThreadPool.createPool(QueueMsgWrapper.ExchageLifecycle.remoteStore.name(), KernelParam.jmsThreadPoolMinSize, KernelParam.jmsThreadPoolMaxSize, Integer.MAX_VALUE);	
		ThreadPool.createPool(QueueMsgWrapper.ExchageLifecycle.feedbackSendStatusAndRoutelog.name(), KernelParam.jmsThreadPoolMinSize, KernelParam.jmsThreadPoolMaxSize, Integer.MAX_VALUE);	
		ThreadPool.createPool(QueueMsgWrapper.ExchageLifecycle.feedbackSendStatusAndRoutelogToTmp.name(), KernelParam.jmsThreadPoolMinSize, KernelParam.jmsThreadPoolMaxSize, Integer.MAX_VALUE);	
		ThreadPool.createPool(QueueMsgWrapper.ExchageLifecycle.feedbackNotifyStatusAndRoutelog.name(), KernelParam.jmsThreadPoolMinSize, KernelParam.jmsThreadPoolMaxSize, Integer.MAX_VALUE);	
		ThreadPool.createPool(QueueMsgWrapper.ExchageLifecycle.feedbackNotifyStatusAndRoutelogTmp.name(), KernelParam.jmsThreadPoolMinSize, KernelParam.jmsThreadPoolMaxSize, Integer.MAX_VALUE);	
		ThreadPool.createPool(QueueMsgWrapper.ExchageLifecycle.remoteStoreTmp.name(), KernelParam.jmsThreadPoolMinSize, KernelParam.jmsThreadPoolMaxSize, Integer.MAX_VALUE);	
		ThreadPool.createPool(QueueMsgWrapper.ExchageLifecycle.gapEntrepot.name(), KernelParam.jmsThreadPoolMinSize, KernelParam.jmsThreadPoolMaxSize, Integer.MAX_VALUE);	
	}
	
	private static final Set<String> idLock = new HashSet<String>();
	private static Set<String> idLocks =Collections.synchronizedSet(idLock);
	/**
	 * 多线程消费
	 * @param receivedData
	 * @throws ServiceException
	 */
	private void startConsumeMsg(final Object receivedData) throws ServiceException{
		final QueueMsgWrapper queueMsgWrapper = (QueueMsgWrapper) receivedData;
		/*SendCmData cmData = (SendCmData)queueMsgWrapper.getData();
		final String  poolName =  queueMsgWrapper.getExchageLifecycle().name();
		final String id = cmData.getId();
		if(poolName.equalsIgnoreCase(QueueMsgWrapper.ExchageLifecycle.feedbackSendStatusAndRoutelog.name())||
				poolName.equalsIgnoreCase(QueueMsgWrapper.ExchageLifecycle.feedbackSendStatusAndRoutelogToTmp.name())){
			while(lockContain(id)){//id被锁住，就等待
				try {
					Thread.sleep(500);//
				} catch (InterruptedException e) {
					//
				}
			}
		}*/
		ThreadPool.execute(queueMsgWrapper.getExchageLifecycle().name(), new Runnable() {
			@Override
			public void run() {
				try {
					consumeMsg(queueMsgWrapper);
				} finally{
					//idLocks.remove(id);
				}
			}
		});
	}
	//id是否被锁住了
	@SuppressWarnings("unused")
	private static boolean lockContain(String id){
		synchronized (idLocks) {
			boolean has = false;
			has = idLocks.contains(id);
			if(!has){
				idLocks.add(id);
			}
			return has;
		}
	}
}
